package cc.eddic.practice.concurrency.impl;

import cc.eddic.practice.concurrency.LogItem;
import cc.eddic.practice.concurrency.LogItemParser;
import cc.eddic.practice.concurrency.LogParser;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

import static cc.eddic.practice.concurrency.ExecutorServiceUtil.shutdown;

@Slf4j
@Component
@Qualifier("LogParserWithCompletableFutureV1Impl")
public class LogParserWithCompletableFutureV1Impl extends LogParserWithFutureImpl implements LogParser {

    @Override
    public Map<String, List<LogItem>> parse(String folderName) {
        Map<String, List<LogItem>> result = new ConcurrentHashMap<>();

        List<CompletableFuture<Model>> futures;
        ExecutorService executorService = Executors.newFixedThreadPool(16);
        try {
            val p = Paths.get(folderName);
            try (val walk = Files.walk(p)) {
                futures = walk.parallel().filter(Files::isRegularFile)
//                        .filter(x -> x.getFileName().toString().startsWith("2023"))
                        .filter(path -> path.getFileName().toString().endsWith(".log"))
                        .map(path -> CompletableFuture.supplyAsync(() -> parseFile(path), executorService))
                        .collect(Collectors.toList());
            }
        } catch (IOException e) {
            log.error("File does not exists : {}", e.getMessage());
            return result;
        }

        if (log.isInfoEnabled()) log.info("{} futures is running", futures.size());
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).thenAccept(aVoid -> {
            if (log.isInfoEnabled()) log.info("all futures is done, build result map");
            futures.stream().parallel().map(CompletableFuture::join)
                    .forEach(model -> result.put(model.getPath().toString(), model.getItems()));
        });

        shutdown(executorService);
        return result;
    }

    public LogParserWithCompletableFutureV1Impl(LogItemParser logItemParser) {
        super(logItemParser);
    }
}
